import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.regression._
import org.apache.spark.mllib.tree.{DecisionTree, RandomForest}
import org.apache.spark.mllib.tree.model.{DecisionTreeModel, RandomForestModel}
import org.apache.spark.rdd.RDD
object RDF {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("mySpark")
    conf.setMaster("local")
    val sc =new SparkContext(conf)

    val rawDate=sc.textFile("/profiledata_06-May-2005/covtype.data")
    rawDate.foreach(println)

//    val data=rawDate.map{line=>
//      val values=line.split(',').map(_.toDouble)
//      val featureVector=Vectors.dense(values.init)
//      val label=values.last-1
//      LabeledPoint(label,featureVector)
//    }
//
//    val Array(trainDate,cvDate,testDate)=data.randomSplit(Array(0.8,0.1,0.1))
//    trainDate.cache();
//    cvDate.cache();
//    testDate.cache();
//
//    def getMetrics(model:DecisionTreeModel,data:RDD[LabeledPoint]):
//      MulticlassMetrics={
//      val predictionsAndLabels=data.map(example=>
//        (model.predict(example.features),example.label)
//      )
//      new MulticlassMetrics(predictionsAndLabels)
//    }
//
//    val model=DecisionTree.trainClassifier(
//      trainDate,7,Map[Int,Int](),"gini",4,100)
//
//    val metrics=getMetrics(model,cvDate)
//
//    (0 until 7).map(
//      cat=>(metrics.precision(cat),metrics.recall(cat))
//    ).foreach(println)
//    println(metrics.accuracy)
//
//    def classProbabilities(data:RDD[LabeledPoint]):Array[Double]={
//      val countsByCategory=data.map(_.label).countByValue()
//      val counts=countsByCategory.toArray.sortBy(_._1).map(_._2)
//      counts.map(_.toDouble/counts.sum)
//    }
//
//    val trainPriorProbabilities=classProbabilities(trainDate)
//    trainPriorProbabilities.foreach(println);
//    val cvPriorProbabilities=classProbabilities(cvDate)
//    cvPriorProbabilities.foreach(println);
//    val testPriorProbabilities=classProbabilities(testDate)
//
//    trainPriorProbabilities.zip(cvPriorProbabilities).map{
//      case (trainProb,cvProb)=>trainProb*cvProb
//    }.sum
//
//    val forest=RandomForest.trainClassifier(
//      trainDate,7,Map(10->4,11->40),20,"auto","entropy",30,300)
//
//    def getForestMetrics(model:RandomForestModel, data:RDD[LabeledPoint]):
//    MulticlassMetrics={
//      val predictionsAndLabels=data.map(example=>
//        (model.predict(example.features),example.label)
//      )
//      new MulticlassMetrics(predictionsAndLabels)
//    }
//    val metrics1=getForestMetrics(forest,cvDate)
//    (0 until 7).map(
//      cat=>(metrics1.precision(cat),metrics1.recall(cat))
//    ).foreach(println)
//    println(metrics1.accuracy)
   }
}
